package com.ktvme.flink13;

import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.ktvme.core.*;

import java.io.File;
import java.util.ArrayList;
import java.util.List;

public class Flink1_13JobServiceImpl extends AbstractFlinkJobService {

    public Flink1_13JobServiceImpl(String baseUrl) {
        super(baseUrl);
    }

    @Override
    public List<JobInfo> getRunningJobs() {
        try (HttpResponse response = HttpRequest.get(baseUrl + "/jobs/overview")
                .execute()) {
            if (!response.isOk()) {
                throw new FlinkApiException("获取作业列表失败: " + response.getStatus());
            }

            JSONObject json = JSONUtil.parseObj(response.body());
            return parseJobList(json.getJSONArray("jobs"), "RUNNING");
        }
    }

    @Override
    public List<JobInfo> getFinishedJobs() {
        try (HttpResponse response = HttpRequest.get(baseUrl + "/jobs/overview")
                .execute()) {
            if (!response.isOk()) {
                throw new FlinkApiException("获取作业列表失败: " + response.getStatus());
            }

            JSONObject json = JSONUtil.parseObj(response.body());
            return parseJobList(json.getJSONArray("jobs"), "FINISHED");
        }
    }

    @Override
    public JobInfo getJobInfo(String jobId) {
        try (HttpResponse response = HttpRequest.get(baseUrl + "/jobs/" + jobId)
                .execute()) {
            if (!response.isOk()) {
                throw new FlinkApiException("获取作业详情失败: " + response.getStatus());
            }

            JSONObject json = JSONUtil.parseObj(response.body());
            return parseJobInfo(json);
        }
    }

    @Override
    public void cancelJob(String jobId) {
        try (HttpResponse response = HttpRequest.get(baseUrl + "/jobs/" + jobId + "/yarn-cancel")
                .execute())
        {
            System.out.println("取消作业响应: " + response);
            if (!response.isOk()) {
                throw new FlinkApiException("取消作业失败: " + response.getStatus());
            }
        }
    }

    @Override
    public void restartFailedJob(String jobId) {
        try (HttpResponse response = HttpRequest.post(baseUrl + "/jobs/" + jobId + "/restart")
                .execute()) {
            if (!response.isOk()) {
                throw new FlinkApiException("重启作业失败: " + response.getStatus());
            }
        }
    }

    @Override
    public List<JarInfo> listJars() {
        try (HttpResponse response = HttpRequest.get(baseUrl + "/jars")
                .execute()) {
            if (!response.isOk()) {
                throw new FlinkApiException("获取JAR包列表失败: " + response.getStatus());
            }

            JSONObject json = JSONUtil.parseObj(response.body());
            List<JarInfo> jars = new ArrayList<>();
            JSONArray files = json.getJSONArray("files");
            
            for (int i = 0; i < files.size(); i++) {
                JSONObject jarJson = files.getJSONObject(i);
                JarInfo jarInfo = new JarInfo();
                jarInfo.setId(jarJson.getStr("id"));
                jarInfo.setName(jarJson.getStr("name"));
                jarInfo.setUploadTime(jarJson.getLong("uploaded"));
                jars.add(jarInfo);
            }
            
            return jars;
        }
    }

    @Override
    public JarInfo uploadJar(File jarFile) {
        try (HttpResponse response = HttpRequest.post(baseUrl + "/jars/upload")
                .form("jarfile", jarFile)  // 注意：参数名应该是"jarfile"而不是"file"
                .execute()) {
            if (!response.isOk()) {
                throw new FlinkApiException("上传JAR文件失败: " + response.getStatus());
            }

            JSONObject json = JSONUtil.parseObj(response.body());
            return parseJarUploadResponse(json);
        }
    }

    @Override
    public void deleteJar(String jarId) {
        try (HttpResponse response = HttpRequest.delete(baseUrl + "/jars/" + jarId)
                .execute()) {
            if (!response.isOk()) {
                throw new FlinkApiException("删除JAR包失败: " + response.getStatus());
            }
        }
    }

    @Override
    public String runJar(RunJarRequest request) {
        return runJar(request.getJarId(), request);
    }

    @Override
    public String runJar(String jarId, RunJarRequest request) {
        try (HttpResponse response = HttpRequest.post(baseUrl + "/jars/" + jarId + "/run")
                .body(JSONUtil.toJsonStr(buildRunParameters(request)))
                .execute()) {
            if (!response.isOk()) {
                throw new FlinkApiException("运行JAR失败: " + response.getStatus());
            }

            JSONObject json = JSONUtil.parseObj(response.body());
            return json.getStr("jobid");
        }
    }

    @Override
    protected List<JobInfo> parseJobList(JSONArray jobsArray, String state) {
        List<JobInfo> jobs = new ArrayList<>();
        for (int i = 0; i < jobsArray.size(); i++) {
            JSONObject jobJson = jobsArray.getJSONObject(i);
            if (state == null || state.equals(jobJson.getStr("state"))) {
                jobs.add(parseJobInfo(jobJson));
            }
        }
        return jobs;
    }

    @Override
    protected JobInfo parseJobInfo(JSONObject json) {
        JobInfo job = new JobInfo();
        job.setJobId(json.getStr("jid"));
        job.setJobName(json.getStr("name"));
        job.setState(json.getStr("state"));
        job.setStartTime(json.getLong("start-time"));
        job.setEndTime(json.getLong("end-time", -1L));
        job.setDuration(json.getLong("duration"));
        job.setLastModification(json.getLong("last-modification", -1L));
        
        if (json.containsKey("tasks")) {
            TaskInfo tasks = json.getJSONObject("tasks").toBean(TaskInfo.class);
            job.setTaskInfo(tasks);
        }
        
        return job;
    }

    @Override
    protected JarInfo parseJarUploadResponse(JSONObject json) {
        JarInfo jarInfo = new JarInfo();
        // Flink REST API 返回的是完整的文件名，需要提取ID部分
        String filename = json.getStr("filename");
        jarInfo.setId(filename);
        jarInfo.setName(filename.substring(filename.lastIndexOf('/') + 1));
        return jarInfo;
    }

    @Override
    protected JSONObject buildRunParameters(RunJarRequest request) {
        JSONObject params = new JSONObject();
        
        if (request.getEntryClass() != null) {
            params.set("entryClass", request.getEntryClass());
        }
        
        if (request.getProgramArgs() != null) {
            params.set("programArgs", request.getProgramArgs());
        }

        if (request.getParallelism() != null) {
            params.set("parallelism", request.getParallelism());
        }

        if (request.getSavepointPath() != null) {
            params.set("savepointPath", request.getSavepointPath());
        }

        if (request.getAllowNonRestoredState() != null) {
            params.set("allowNonRestoredState", request.getAllowNonRestoredState());
        }

        return params;
    }
}
